from typing import List, Dict, Any, Type, Optional
import pandas as pd
from fastapi import UploadFile
from io import BytesIO
from datetime import datetime
from pydantic import BaseModel


class ExcelUtil:
    @staticmethod
    def create_import_template(
            columns: List[Dict[str, str]],
            example_data: Optional[List[Dict[str, Any]]] = None,
            sheet_name: str = "导入模板"
    ) -> BytesIO:
        """创建通用导入模板"""
        # 构建表头
        header = []
        for col in columns:
            title = col["title"]
            if col.get("required", False):
                title += "(*)"
            header.append(title)

        # 创建数据
        if not example_data:
            data = []
        else:
            data = []
            for row in example_data:
                formatted_row = []
                for col in columns:
                    formatted_row.append(row.get(col["field"], ""))
                data.append(formatted_row)

        # 创建DataFrame
        df = pd.DataFrame(data, columns=header)

        # 写入Excel
        output = BytesIO()
        writer = pd.ExcelWriter(
            output,
            engine='xlsxwriter',
            engine_kwargs={'options': {'nan_inf_to_errors': True}}
        )

        df.to_excel(writer, sheet_name=sheet_name, index=False)
        writer.close()
        output.seek(0)

        return output

    @staticmethod
    def parse_import_data(
            file: BytesIO,
            columns: List[Dict[str, str]],
            schema_cls: Optional[Type[BaseModel]] = None
    ) -> List[Dict[str, Any]]:
        """解析导入的Excel文件"""
        try:
            df = pd.read_excel(file, engine='openpyxl')
        except Exception as e:
            raise ValueError(f"无法读取Excel文件: {str(e)}")

        # 验证必填字段
        required_columns = [f"{col['title']}(*)" for col in columns if col.get("required", False)]
        for column in required_columns:
            if column not in df.columns:
                raise ValueError(f"缺少必填字段: {column}")

        # 构建字段映射
        field_map = {}
        for col in columns:
            title = col["title"] + ("(*)" if col.get("required", False) else "")
            field_map[title] = col["field"]

        # 转换数据
        result = []
        for idx, row in df.iterrows():
            item = {}
            for title, value in row.items():
                # 清理标题
                clean_title = title.replace("(*)", "")
                field = None
                for col in columns:
                    if col["title"] == clean_title:
                        field = col["field"]
                        break

                if field and pd.notna(value):
                    # 数据类型处理
                    if isinstance(value, (float, int)):
                        if float(value).is_integer():
                            value = int(value)
                    elif isinstance(value, str):
                        value = value.strip()
                    item[field] = value

            if schema_cls:
                try:
                    item = schema_cls(**item).dict(exclude_unset=True)
                except Exception as e:
                    raise ValueError(f"第{idx + 1}行数据验证失败: {str(e)}")

            result.append(item)

        return result

    @staticmethod
    async def export_data(
            data: List[Any],
            columns: List[Dict[str, str]],
            sheet_name: str = "数据导出"
    ) -> BytesIO:
        """导出数据到Excel"""
        # 转换数据
        rows = []
        for item in data:
            row = {}
            for col in columns:
                field = col["field"]
                # 处理嵌套字段
                value = item
                for key in field.split('.'):
                    if hasattr(value, key):
                        value = getattr(value, key)
                    elif isinstance(value, dict):
                        value = value.get(key)
                    else:
                        value = None
                    if value is None:
                        break

                # 处理特殊类型
                if isinstance(value, datetime):
                    value = value.strftime('%Y-%m-%d %H:%M:%S')
                elif value is None:
                    value = ""
                # 处理枚举类型
                elif hasattr(value, 'value'):  # 检查是否是枚举类型
                    value = str(value.value)  # 确保转换为字符串

                # 使用格式化器（如果有）
                formatter = col.get('formatter')
                if formatter and callable(formatter):
                    value = formatter(value)  # 传递字符串值给格式化函数

                row[col["title"]] = value
            rows.append(row)

        # 创建DataFrame
        df = pd.DataFrame(rows)

        # 数据清理
        for column in df.columns:
            df[column] = df[column].apply(lambda x: '' if pd.isna(x) else str(x))

        # 创建输出缓冲区
        output = BytesIO()

        try:
            # 使用xlsxwriter引擎
            writer = pd.ExcelWriter(
                output,
                engine='xlsxwriter',
                engine_kwargs={'options': {'nan_inf_to_errors': True}}
            )

            # 写入数据
            df.to_excel(writer, sheet_name=sheet_name, index=False)

            # 获取workbook和worksheet对象
            workbook = writer.book
            worksheet = writer.sheets[sheet_name]

            # 设置列宽
            for i, col in enumerate(df.columns):
                max_length = max(
                    df[col].astype(str).str.len().max(),
                    len(str(col))
                )
                worksheet.set_column(i, i, max_length + 2)

            # 关闭writer
            writer.close()

        except Exception as e:
            raise ValueError(f"导出Excel失败: {str(e)}")

        output.seek(0)
        return output